/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.rocketmq.client.latency;

import org.apache.rocketmq.client.impl.producer.TopicPublishInfo;
import org.apache.rocketmq.client.log.ClientLogger;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.common.message.MessageQueue;

/**
 * 消息失败策略，延迟实现的门面类。
 */
public class MQFaultStrategy {
    private final static InternalLogger log = ClientLogger.getLog();
    private final LatencyFaultTolerance<String> latencyFaultTolerance = new LatencyFaultToleranceImpl();

    private boolean sendLatencyFaultEnable = false;

    /**
     * 消息发送的延迟时间 和该数组进行对比。 从该数组的尾部开始和消息发送的延迟时间进行对比，找到第一个比延迟时间小的值。
     * 并获取这个数组的索引。
     */
    private long[] latencyMax = {50L, 100L, 550L, 1000L, 2000L, 3000L, 15000L};
    /**
     * 根据latencyMax 的索引 获取规避broker对应的时间。
     */
    private long[] notAvailableDuration = {0L, 0L, 30000L, 60000L, 120000L, 180000L, 600000L};

    public long[] getNotAvailableDuration() {
        return notAvailableDuration;
    }

    public void setNotAvailableDuration(final long[] notAvailableDuration) {
        this.notAvailableDuration = notAvailableDuration;
    }

    public long[] getLatencyMax() {
        return latencyMax;
    }

    public void setLatencyMax(final long[] latencyMax) {
        this.latencyMax = latencyMax;
    }

    public boolean isSendLatencyFaultEnable() {
        return sendLatencyFaultEnable;
    }

    public void setSendLatencyFaultEnable(final boolean sendLatencyFaultEnable) {
        this.sendLatencyFaultEnable = sendLatencyFaultEnable;
    }

    /**
     * 选择一个消息队列
     *
     * @param tpInfo         消息路由信息
     * @param lastBrokerName 最后一个发送到broker的名字
     * @return
     */
    public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
        //sendLatencyFaultEnable==false 默认不启用broker故障延迟机制 true 就是指启用broker故障延迟机制。
        if (this.sendLatencyFaultEnable) {
            try {
                //每选择一次消息队列该值会增加1。
                int index = tpInfo.getSendWhichQueue().getAndIncrement();
                //遍历该主题的消息队列值
                for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
                    //取模
                    int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
                    if (pos < 0)
                        pos = 0;
                    //根据取模 选择队列
                    MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
                    //根据队列的borkerName去验证该消息队列是否可以使用。
                    if (latencyFaultTolerance.isAvailable(mq.getBrokerName())) {
                        //lastBrokerName ==null 属于非失败重试的。直接返回队列
                        //如果lastBrokerName！=null 那么属于失败重试的，如果选择的队列和上次是一样的则返回。
                        if (null == lastBrokerName || mq.getBrokerName().equals(lastBrokerName))
                            return mq;
                    }
                }

                //从规避的broker中尝试获取一个可用的
                final String notBestBroker = latencyFaultTolerance.pickOneAtLeast();
                //代表这个之前故障的broker已经恢复了
                int writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);
                if (writeQueueNums > 0) {
                    //选择一个消息队列
                    final MessageQueue mq = tpInfo.selectOneMessageQueue();
                    if (notBestBroker != null) {
                        //如果notBestBroker！=null 但是有可写入的队列，代表故障broker已经恢复了
                        //将取到的队列信息设置为取到的broker
                        mq.setBrokerName(notBestBroker);
                        //队列重置
                        mq.setQueueId(tpInfo.getSendWhichQueue().getAndIncrement() % writeQueueNums);
                    }
                    return mq;
                } else {
                    //这个broker没有写队列就移除。※ 有点疑惑：为什么没有写队列就移除？应该是没有写对了 消息都发不了吧
                    latencyFaultTolerance.remove(notBestBroker);
                }
            } catch (Exception e) {
                log.error("Error occurred when selecting message queue", e);
            }

            return tpInfo.selectOneMessageQueue();
        }

        return tpInfo.selectOneMessageQueue(lastBrokerName);
    }

    public void updateFaultItem(final String brokerName, final long currentLatency, boolean isolation) {
        //必须启用了broker故障延迟机制 才会进入修改broker失败条目的代码
        if (this.sendLatencyFaultEnable) {
            //根据isolation计算是使用默认的30S规避broker的时间 还是用消息延迟发送的时间
            long duration = computeNotAvailableDuration(isolation ? 30000 : currentLatency);
            this.latencyFaultTolerance.updateFaultItem(brokerName, currentLatency, duration);
        }
    }

    /**
     * 根据本次消息发送延迟，从latencyMax的尾部向前找到第一个比currentLatency 小的
     * 索引
     *
     * @param currentLatency 本次消息发送延迟
     * @return
     */
    private long computeNotAvailableDuration(final long currentLatency) {
        for (int i = latencyMax.length - 1; i >= 0; i--) {
            if (currentLatency >= latencyMax[i])
                //然后根据索引从notAvailableDuration数组中取出对应的时间，在这个时长内 broker将设置为不可用
                return this.notAvailableDuration[i];
        }

        //如果找不到这个索引就返回初始位置。
        return 0;
    }
}
